package com.hdu.dwh.dao;

import com.hdu.dwh.utils.MinioUtil;
import com.hdu.dwh.utils.StatementUtils;

import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.catalog.ResolvedSchema;
import org.springframework.stereotype.Component;

import com.hdu.dwh.pojos.PaimonTag;
import com.hdu.dwh.pojos.SQLBuilder.SQLCreateTableBuilder;
import com.hdu.dwh.pojos.SQLBuilder.SQLSelectBuilder;
import com.hdu.dwh.pojos.source_access_pojos.IColumn;
import com.hdu.dwh.properties.FlinkProperties;

import lombok.Data;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import static org.apache.flink.table.api.Expressions.$;

import java.util.Map;
import java.util.Set;


@Slf4j
@Component
@RequiredArgsConstructor
@Data
public class PaimonBaseDao {

    private final TableEnvironment tableEnvironment;

    private final FlinkBaseDao flinkBaseDao;
    private final FlinkProperties flinkProperties;
    private final MinioUtil minioUtil;

    /**
     * 查看 paimon里 的 所有 数据库
     * @return
     */
    public String[] showDatabase() {
        return flinkBaseDao.showDatabase(flinkProperties.getCatalog());
    }

    /**
     * 查看 paimon里 的 某数据库里的 所有 数据表
     * @return
     */
    public String[] showTable(String databaseName){
        return flinkBaseDao.showTable(flinkProperties.getCatalog(), databaseName);
    }


    /**
     * 检查 paimon 里 数据库 是否存在
     * @param databaseName
     * @return
     */
    public boolean checkDatabaseExists(String databaseName) {
        return flinkBaseDao.checkDatabaseExists(flinkProperties.getCatalog(),databaseName);
    }


    /**
     * 检查 paimon 里 某数据库 的 数据表是否存在
     * @param databaseName
     * @return
     */
    public boolean checkTableExists(String databaseName, String tableName) {
        return flinkBaseDao.checkTableExists(flinkProperties.getCatalog(), databaseName, tableName);
    }


    /**
     * 在paimon里创建数据库
     * @param databaseName
     */
    public boolean createDatabase(String databaseName) {
        return flinkBaseDao.createDatabase(flinkProperties.getCatalog(), databaseName);
    }


    /**
     * 在paimon里删除数据库
     * @param databaseName
     */
    public boolean deleteDatabase(String databaseName) {
        return flinkBaseDao.deleteDatabase(flinkProperties.getCatalog(),databaseName);
    }

    public void alterDatabaseName(String originName, String newName) {
        
    }

    /**
     * 在paimon里删除表
     * @param databaseName
     * @param tableName
     */
    public void deleteTable(String databaseName, String tableName) {
        this.flinkBaseDao.deleteTable(this.flinkProperties.getCatalog(), databaseName, tableName);
    }

    /**
     * 给当前数据库添加tag
     * @param databaseName
     * @param tableName
     * @param tagName
     * @param snapshot
     */
    public void createTagAtLatestSnapshot(String databaseName, String tableName, String tagName) {
        this.tableEnvironment.executeSql("CALL sys.create_tag('%s.%s', '%s')".formatted(databaseName, tableName, tagName));
    }

    /**
     * 退回到指定的tag版本
     * @param databaseName
     * @param tableName
     * @param tagName
     */
    public void rollBackToTag(String databaseName, String tableName, String tagName) {
        this.tableEnvironment.executeSql("CALL sys.rollback_to('%s.%s', '%s')".formatted(databaseName, tableName, tagName));
    }

    /**
     * 删除指定的tag
     * @param databaseName
     * @param tableName
     * @param tagName
     */
    public void deleteTag(String databaseName, String tableName, String tagName) {
        this.tableEnvironment.executeSql("CALL sys.delete_tag('%s.%s', '%s')".formatted(databaseName, tableName, tagName));
    }

    /**
     * 展示表的所有tag
     * @param databaseName
     * @param tableName
     */
    public Set<PaimonTag> showTags(String databaseName, String tableName) {
        Table tagTable = this.tableEnvironment.sqlQuery("SELECT * FROM `%s`.`%s$tags`".formatted(databaseName, tableName));
        tagTable.select($("tag_name"));
        TableResult tagResult = tagTable.execute();
        Set<PaimonTag> tags_name = new java.util.HashSet<>();
    tagResult.collect().forEachRemaining(row -> {
        PaimonTag tag = new PaimonTag();
        Object tagNameField = row.getField("tag_name");
        tag.setTagName(tagNameField != null ? tagNameField.toString() : null);
        Object commitTimeField = row.getField("commit_time");
        tag.setCommit_time(commitTimeField != null ? commitTimeField.toString() : null);
        Object recordCountField = row.getField("record_count");
        tag.setRecord_count(recordCountField != null ? recordCountField.toString() : null);
        tags_name.add(tag);

    });
        return tags_name;
    }

    /**
     * 执行sql语句,返回jobid
     * @param sql
     * @return jobid {@link String} 
     */
    public String executeSQL(String sql) {
        log.info("Executing SQL: \n{}\n", sql);
        // this.tableEnvironment.explainSql(sql);
        TableResult result = this.tableEnvironment.executeSql(sql);
        if(!result.getJobClient().isPresent()) {
            log.info("SQL executed successfully, but no job was submitted.");
            return "执行成功，但没有提交job";
        }
        return result.getJobClient().get().getJobID().toString();
    }

    public void deleteTable(Long tableId) {
        
    }

    public void createPaimonTable(String databaseName, String tableName, Set<IColumn> columns, Set<String> primaryKeys, Set<String> partitionKeys, Map<String, String> options) {
        if(this.checkTableExists(databaseName, tableName)) {
            throw new RuntimeException("Table %s.%s already exist".formatted(databaseName, tableName));
        }
        SQLCreateTableBuilder builder = new SQLCreateTableBuilder(databaseName, tableName, columns)
            .withPrimaryKeys(primaryKeys)
            .withPartitionKeys(partitionKeys)
            .withTableOptions(options);
        this.executeSQL(builder.build());
    }

    public void createPaimonTable(String databaseName, String tableName, Set<IColumn> columns) {
        this.createPaimonTable(databaseName, tableName, columns, null, null, null);
    }

}
